package com.rpt

import com.beans.AdLog
import com.dmp.RptUtils
import com.utils.JedisPools
import org.apache.commons.lang.StringUtils
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * ClassName:MediaAnalyseRptV2
  * Package:com.rpt
  * Desciption:媒体报表分析
  *
  * @date:2019 /8/24 0:51
  * @author:17611219021 @sina.cn
  */
object MediaAnalyseRptV2 {
  def main(args: Array[String]): Unit = {
    if(args.length !=2){
      println(
        """
          |com.dmp.fileToParquet
          |参数：
          | inputpath
          | outputPath
        """.stripMargin)
      sys.exit()
    }
    val sparkConf = new SparkConf()
    sparkConf.setAppName(s"${this.getClass.getSimpleName}")
    sparkConf.setMaster("local[*]")
    //指定keyo序列化器与snappy格式，默认是gzip格式
    sparkConf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
    sparkConf.set("spark.sql.parquet.compression.codec","snappy")
    val sc: SparkContext = new SparkContext(sparkConf)
    val Array(inputpath,outputPath)= args

    val configuration = sc.hadoopConfiguration
    val fs = FileSystem.get(configuration)
    val path = new Path(outputPath)
    if(fs.exists(path)){
      fs.delete(path,true)
    }


    sc.textFile(inputpath).map(_.split(",",-1)).filter(_.length>=3).
      map(AdLog(_)).mapPartitions(itr =>{
      val jedis=JedisPools.getJedis()
      var pairResult= new collection.mutable.ListBuffer[(String,List[Double])]()
      //遍历分区的所有数据，将结果放到listBuffer
      itr.foreach(adlog=>{
        var newappname=adlog.appname
        if(StringUtils.isEmpty(newappname)){
          newappname=jedis.get(adlog.appid)
        }
        //原始请求、有效请求、广告请求
        val listReq = RptUtils.calculateReq(adlog.requestmode,adlog.processnode)

        //参与竞价、竞价成功、消费、成本
        val listRtb = RptUtils.calculateRtb(adlog.iseffective,adlog.isbiding,adlog.isbid,adlog.adorderid,adlog.iswin,adlog.adpayment,adlog.winprice)

        //广告展示、广告点击
        val listShowOrClick = RptUtils.calculateShowOrClick(adlog.requestmode,adlog.iseffective)
        //(newappname,listReq++listRtb++listShowOrClick)
        pairResult+=((newappname,listReq++listRtb++listShowOrClick))
      })
      jedis.close()
      //将buffer转为迭代器
      pairResult.toIterator
    }).reduceByKey((list1,list2) =>{
      list1.zip(list2).map(t=>{t._1+t._2})
    }).saveAsTextFile(outputPath)
      sc.stop()

  }

}
